using System;
using System.Collections.Generic;
using System.Linq;
using CoreLibrary.Collection;
using CoreLibrary.Concurrent;
using CoreLibrary.Messaging.Sql;
using CoreLibrary.Messaging.Store;
using CoreLibrary.Scheduler;
using CoreLibrary.Service;
using NLog;

namespace CoreLibrary.Messaging {

    public class MessageQueue : ServiceBase, IMessageQueue {

        public ISchedulerService SchedulerService { get; set; }

        public IMessageStore MessageStore { get; set; }

        public ExceptionHandler UncaughtExceptionHandler { get; set; }
        public int MaxThreadCount { get; set; }

        private readonly CopyOnWriteList<IMessageHandler> _messageHandlers = new CopyOnWriteList<IMessageHandler>();
        private AsyncBlockingTaskQueue _taskTaskQueue;

        private MessageQueueStatus _queueStatus = MessageQueueStatus.Stopped;

        public MessageQueue() {
            MaxThreadCount = 20;
        }

        protected override void OnInit() {
            Asserts.NotNull(SchedulerService, "Scheduler service is not in place");
            Asserts.NotNull(MessageStore, "Message store is not in place");
            Asserts.True(MaxThreadCount > 0, "Max thread count must be greater than 0");

            _taskTaskQueue = new AsyncBlockingTaskQueue(MaxThreadCount) {
                Executor = ProcessMessage,
                ErrorHandler = ProcessError
            };
        }

        protected override void OnStart() {
            List<Message> oldMessages;

            lock (this) {
                if (_queueStatus == MessageQueueStatus.Started) {
                    return;
                }
                /**
                 * Old message would be reloaded but not executed until task queue starts. 
                 */
                oldMessages = ReloadOldMessages();
                _taskTaskQueue.Start();
                _queueStatus = MessageQueueStatus.Started;
            }

            SchedulerService.Schedule(() => {
                var now = DateTime.Now;
                foreach (var message in oldMessages) {
                    var span = message.DateScheduled.Subtract(now);
                    if (message.Status != MessageStatus.Enqueued) {
                        UpdateMessageToEnqueued(message);
                    }
                    if (span.TotalMinutes < 3) {
                        SchedulerService.Schedule(EnqueueMessageIfAbsent, message, TimeSpan.FromMinutes(1));
                    } 
                    else {
                        SchedulerService.Schedule(EnqueueMessageIfAbsent, message, span);
                    }
                }
                LogManager.GetCurrentClassLogger().Info("Enqueued " + oldMessages.Count + " old messages.");
            }, TimeSpan.FromSeconds(30));
        }

        protected override void OnStop() {
            lock (this) {
                if (_queueStatus == MessageQueueStatus.Stopping ||
                    _queueStatus == MessageQueueStatus.Stopped) {
                    return;
                }
                _taskTaskQueue.Stop();
                _queueStatus = MessageQueueStatus.Stopping;
            }
        }

        protected override void OnDestroy() {
        }

        public void InsertMessageHandler(IMessageHandler handler) {
            if (!_messageHandlers.Contains(handler)) {
                 _messageHandlers.Add(handler);
            }
        }

        public void RemoveMessageHandler(IMessageHandler handler) {
            _messageHandlers.Remove(handler);
        }

        public SendMessageBatchResponse SendMessageBatch(SendMessageBatchRequest request) {
            throw new NotImplementedException();
        }

        public List<IMessageHandler> ListMessageHandlers() {
            return _messageHandlers.GetCopyForRead();
        }

        public MessageQueueStatus GetQueueState() {
            lock (this) {
                if (_queueStatus == MessageQueueStatus.Stopping &&
                    _taskTaskQueue.CountActive() == 0) {
                    _queueStatus = MessageQueueStatus.Stopped;
                }
                return _queueStatus;
            }
        }

        public int GetMessageCount() {
            return _taskTaskQueue.Count();
        }

        public int GetMaxActiveMessageCount() {
            return _taskTaskQueue.MaxThread;
        }

        public int GetProcessingMessageCount() {
            return _taskTaskQueue.CountActive();
        }

        public int GetAwaitingMessageCount() {
            return _taskTaskQueue.CountInactive();
        }

        public List<Message> ListMessages(int offset, int max) {
            var tasks = _taskTaskQueue.Get(offset, max);
            return tasks.Cast<Message>().ToList();
        }

        public SendMessageResponse SendMessage(SendMessageRequest request) {
            lock (this) {
                var message = new Message {
                    Id            = Guid.NewGuid().ToString(),
                    ApplicationId = request.ApplicationId,
                    Data          = request.Body,
                    Status        = MessageStatus.Enqueued,
                    Fault         = "",
                    DateCreated   = request.DateCreated,
                    DateScheduled = request.DateScheduled,
                    Priority      = request.Priority
                };

                try {
                    MessageStore.InsertMessage(message);
                } 
                catch (Exception e) {
                    LogManager.GetCurrentClassLogger().ErrorException("Storing message failed: " + e.Message, e);
                    return new SendMessageResponse {
                           Success      = false,
                           MessageId    = null,
                           ErrorMessage = "Storing message failed: " + e.Message,
                    };
                }

                try {
                    TimeSpan span = message.DateScheduled.Subtract(DateTime.Now);
                    if (span.TotalMinutes < 3) {
                        EnqueueMessage(message);
                    } else {
                        SchedulerService.Schedule(EnqueueMessage, message, span);
                    }
                } catch (Exception e) {
                    LogManager.GetCurrentClassLogger().ErrorException("Scheduling message failed: " + e.Message, e);
                    return new SendMessageResponse {
                           Success      = false,
                           MessageId    = null,
                           ErrorMessage = "Scheduling message failed: " + e.Message,
                    };
                }

                return new SendMessageResponse {
                    Success      = true,
                    MessageId    = message.Id,
                    ErrorMessage = ""
                };
            }
        }

        public Message RetryMessage(string messageId, string errorMsg) {
            var message = MessageStore.GetMessage(messageId);
            if (message == null) {
                throw new Exception("Message#" + messageId + " doesn't exist or has been deleted.");
            }

            var currentStatus = message.Status;
            if (currentStatus == MessageStatus.Enqueued) {
                throw new Exception("Message#" + messageId + " is already in queue.");
            }
            if (currentStatus == MessageStatus.Processing) {
                throw new Exception("Message#" + messageId + " is under processing.");
            }
            if (currentStatus == MessageStatus.Processed) {
                throw new Exception("Message#" + messageId + " had been processed.");
            }

            if (CompareAndUpdateMessageToRetry(message, currentStatus, errorMsg)) {
                _taskTaskQueue.EnqueueIfAbsent(message, GetTaskPriority(message.Priority));
            } 
            else {
                throw new Exception("Message#" + messageId + " is being updated by another thread.");
            }
            return message;
        }

        public Message RemoveMessage(string messageId) {
            var message = MessageStore.GetMessage(messageId);
            if (message == null) {
                return null;
            }

            MessageStore.RemoveMessage(message.Id);
            _taskTaskQueue.Remove(message);
            return message;
        }

        private static AsyncBlockingTaskQueue.Priority GetTaskPriority(MessagePriority priority) {
            if (priority == MessagePriority.High) {
                return AsyncBlockingTaskQueue.Priority.High;
            }
            return AsyncBlockingTaskQueue.Priority.Normal;
        }

        public List<Message> ReloadOldMessages() {
            var messages = new List<Message>();
            try {
                var query = new ListMessageQuery {
                    Asc = false,
                    Offset = 0,
                    Limit = 10000000,
                    MessageStatuses = new[] { (int)MessageStatus.Enqueued, (int)MessageStatus.Processing },
                };

                var count = 0;
                MessageStore.ListMessages(query, message => {
                    messages.Add(message);
                    ++count;
                });
                LogManager.GetCurrentClassLogger().Info("Reloaded " + count + " old messages");
            } 
            catch (Exception e) {
                LogManager.GetCurrentClassLogger().ErrorException("Reloading old messages failed: " + e.Message, e);
            }

            return messages;
        }

        private void EnqueueMessage(object task) {
            var message = (Message)task;
            _taskTaskQueue.Enqueue(message, GetTaskPriority(message.Priority));
        }

        private void EnqueueMessageIfAbsent(object task) {
            var message = (Message)task;
            _taskTaskQueue.EnqueueIfAbsent(message, GetTaskPriority(message.Priority));
        }

        private void ProcessMessage(object task, AsyncBlockingTaskQueue.CompletionCallback completionCallback) {
            var message = (Message)task;

            try {
                bool updated = UpdateMessageToProcessing(message);
                /**
                 * Update fails usually when the message has been deleted from database. in which case
                 * it just returns.
                 */
                if (!updated) {
                    return;
                }

                MessageHandlingCallback callback = null;
                foreach (var messageHandler in _messageHandlers) {
                    if (messageHandler.CanHandleMessage(message)) {
                        callback = new MessageHandlingCallback(this, message);
                        messageHandler.HandleMessage(message, callback);
                        break;
                    }
                }

                if (callback == null) {
                    /**
                     * when message queue is stopped or stopping, some handlers
                     * may have been stopped or destroyed. Don't move message to
                     * pending state so that the message will be processed again
                     * when the queue is back running.
                     */
                    if (_queueStatus != MessageQueueStatus.Stopping &&
                        _queueStatus != MessageQueueStatus.Stopped) {
                        UpdateMessageToPending(message);
                    }
                }
            } catch (Exception error) {
                ProcessError(message, error);
                if (UncaughtExceptionHandler != null) {
                    try {
                        UncaughtExceptionHandler(error);
                    } catch (Exception failure) {
                        LogManager.GetCurrentClassLogger().ErrorException("Message queue uncaught exception handling failed: " + failure.Message, failure);
                    }
                } else {
                    LogManager.GetCurrentClassLogger().ErrorException("Processing message failed: " + error.Message, error);
                }
            } finally {
                completionCallback();
            }
        }

        private void ProcessError(object task, Exception error) {
            var message = (Message)task;
            /**
             * when message queue is stopped or stopping, message processing
             * may fail. Don't move message to failed state so that the message 
             * will be processed again when the queue is back running.
             */
            if (_queueStatus == MessageQueueStatus.Stopping ||
                _queueStatus == MessageQueueStatus.Stopped) {
                return;
            }
            UpdateMessageToFailed(message, error);
        }

        private bool UpdateMessageToEnqueued(Message message) {
            message.Status = MessageStatus.Enqueued;
            return MessageStore.UpdateMessageStatus(message.Id, message.Status, message.Fault, message.Retry, message.DateScheduled);
        }

        private bool UpdateMessageToProcessing(Message message) {
            message.Status = MessageStatus.Processing;
            return MessageStore.UpdateMessageStatus(message.Id, message.Status, message.Fault, message.Retry, message.DateScheduled);
        }

        private bool UpdateMessageToFailed(Message message, string errorMsg) {
            message.Status = MessageStatus.Failed;
            message.Fault  = StringUtil.SubString(errorMsg, 0, 1000);
            return MessageStore.UpdateMessageStatus(message.Id, message.Status, message.Fault, message.Retry, message.DateScheduled);
        }

        private bool UpdateMessageToFailed(Message message, Exception error) {
            return UpdateMessageToFailed(message, Util.GetString(error.Message));
        }

        private bool UpdateMessageToPending(Message message) {
            message.Status = MessageStatus.Pending;
            return MessageStore.UpdateMessageStatus(message.Id, message.Status, message.Fault, message.Retry, message.DateScheduled);
        }

        private bool UpdateMessageToProcessed(Message message) {
            message.Status = MessageStatus.Processed;
            MessageStore.UpdateMessageStatus(message.Id, message.Status, message.Fault, message.Retry, message.DateScheduled);
            return MessageStore.RemoveMessage(message.Id);
        }

        private bool UpdateMessageToRetry(Message message, string errorMsg, DateTime dateScheduled) {
            message.Status = MessageStatus.Enqueued;
            message.Fault = errorMsg;
            message.Retry += 1;
            message.DateScheduled = dateScheduled;
            return MessageStore.UpdateMessageStatus(message.Id, message.Status, message.Fault, message.Retry, message.DateScheduled);
        }

        private bool CompareAndUpdateMessageToRetry(Message message, MessageStatus expectedStatus, string errorMsg) {
            const MessageStatus newStatus = MessageStatus.Enqueued;
            var msg = StringUtil.SubString(errorMsg, 0, 1000);

            bool updated = MessageStore.CompareAndUpdateMessageStatus(
                message.Id,
                expectedStatus,
                newStatus,
                msg,
                message.Retry + 1,
                message.DateScheduled
            );

            if (updated) {
                message.Status = newStatus;
                message.Fault = msg;
                message.Retry += 1;
            }

            return updated;
        }

        private class MessageHandlingCallback : IMessageProcessingCallback {

            private readonly MessageQueue _messageQueue;
            private readonly Message _message;

            public MessageHandlingCallback(MessageQueue messageQueue, Message message) {
                _messageQueue = messageQueue;
                _message = message;
            }

            public void OnSuccess() {
                _messageQueue.UpdateMessageToProcessed(_message);
            }

            public void OnFailure(string errorMsg) {
                _messageQueue.UpdateMessageToFailed(_message, errorMsg);
            }

            public void OnRetry(string errorMsg, TimeSpan delay) {
                _messageQueue.UpdateMessageToRetry(_message, errorMsg, DateTime.Now.Add(delay));
                _messageQueue.SchedulerService.Schedule(_messageQueue.EnqueueMessage, _message, delay);
            }
        }

    }
}